diff --git a/MANIFEST.in b/MANIFEST.in
index 4c357df7..6605f276 100644
--- a/MANIFEST.in
+++ b/MANIFEST.in
@@ -1,18 +1,16 @@
include Makefile
include pytest.ini
include README.md
include requirements.txt
include requirements-swh.txt
include requirements-test.txt
include tox.ini
include version.txt
recursive-include swh/web/assets *
recursive-include swh/web/static *
recursive-include swh/web/templates *
-include swh/web/tests/browse/views/data/swh-logo.png
-include swh/web/tests/browse/views/data/iso-8859-1_encoded_content
recursive-include swh/web/tests/resources *
include package-lock.json
include package.json
include yarn.lock
diff --git a/swh/web/browse/utils.py b/swh/web/browse/utils.py
index 82201c39..50b753dc 100644
--- a/swh/web/browse/utils.py
+++ b/swh/web/browse/utils.py
@@ -1,1123 +1,1129 @@
-# Copyright (C) 2017-2018 The Software Heritage developers
+# Copyright (C) 2017-2019 The Software Heritage developers
# See the AUTHORS file at the top-level directory of this distribution
# License: GNU Affero General Public License version 3, or any later version
# See top-level LICENSE file for more information
import base64
from collections import defaultdict
import magic
import pypandoc
import stat
import textwrap
from django.core.cache import cache
from django.utils.safestring import mark_safe
from importlib import reload
from swh.model.identifiers import persistent_identifier
from swh.web.common import highlightjs, service
from swh.web.common.exc import NotFoundExc, http_status_code_message
from swh.web.common.origin_visits import get_origin_visit
from swh.web.common.utils import (
reverse, format_utc_iso_date, get_swh_persistent_id,
swh_object_icons
)
from swh.web.config import get_config
def get_directory_entries(sha1_git):
"""Function that retrieves the content of a directory
from the archive.
The directories entries are first sorted in lexicographical order.
Sub-directories and regular files are then extracted.
Args:
sha1_git: sha1_git identifier of the directory
Returns:
A tuple whose first member corresponds to the sub-directories list
and second member the regular files list
Raises:
NotFoundExc if the directory is not found
"""
cache_entry_id = 'directory_entries_%s' % sha1_git
cache_entry = cache.get(cache_entry_id)
if cache_entry:
return cache_entry
entries = list(service.lookup_directory(sha1_git))
for e in entries:
e['perms'] = stat.filemode(e['perms'])
if e['type'] == 'rev':
# modify dir entry name to explicitly show it points
# to a revision
e['name'] = '%s @ %s' % (e['name'], e['target'][:7])
dirs = [e for e in entries if e['type'] in ('dir', 'rev')]
files = [e for e in entries if e['type'] == 'file']
dirs = sorted(dirs, key=lambda d: d['name'])
files = sorted(files, key=lambda f: f['name'])
cache.set(cache_entry_id, (dirs, files))
return dirs, files
def get_mimetype_and_encoding_for_content(content):
"""Function that returns the mime type and the encoding associated to
a content buffer using the magic module under the hood.
Args:
content (bytes): a content buffer
Returns:
A tuple (mimetype, encoding), for instance ('text/plain', 'us-ascii'),
associated to the provided content.
"""
while True:
try:
magic_result = magic.detect_from_content(content)
mime_type = magic_result.mime_type
encoding = magic_result.encoding
break
except Exception:
# workaround an issue with the magic module who can fail
# if detect_from_content is called multiple times in
# a short amount of time
reload(magic)
return mime_type, encoding
# maximum authorized content size in bytes for HTML display
# with code highlighting
content_display_max_size = get_config()['content_display_max_size']
snapshot_content_max_size = get_config()['snapshot_content_max_size']
+def _reencode_content(mimetype, encoding, content_data):
+ # encode textual content to utf-8 if needed
+ if mimetype.startswith('text/'):
+ # probably a malformed UTF-8 content, re-encode it
+ # by replacing invalid chars with a substitution one
+ if encoding == 'unknown-8bit':
+ content_data = content_data.decode('utf-8', 'replace')\
+ .encode('utf-8')
+ elif 'ascii' not in encoding and encoding not in ['utf-8', 'binary']:
+ content_data = content_data.decode(encoding, 'replace')\
+ .encode('utf-8')
+ elif mimetype.startswith('application/octet-stream'):
+ # file may detect a text content as binary
+ # so try to decode it for display
+ encodings = ['us-ascii']
+ encodings += ['iso-8859-%s' % i for i in range(1, 17)]
+ for encoding in encodings:
+ try:
+ content_data = content_data.decode(encoding)\
+ .encode('utf-8')
+ except Exception:
+ pass
+ else:
+ # ensure display in content view
+ mimetype = 'text/plain'
+ break
+ return mimetype, content_data
+
+
def request_content(query_string, max_size=content_display_max_size,
raise_if_unavailable=True, reencode=True):
"""Function that retrieves a content from the archive.
Raw bytes content is first retrieved, then the content mime type.
If the mime type is not stored in the archive, it will be computed
using Python magic module.
Args:
query_string: a string of the form "[ALGO_HASH:]HASH" where
optional ALGO_HASH can be either ``sha1``, ``sha1_git``,
``sha256``, or ``blake2s256`` (default to ``sha1``) and HASH
the hexadecimal representation of the hash value
max_size: the maximum size for a content to retrieve (default to 1MB,
no size limit if None)
Returns:
A tuple whose first member corresponds to the content raw bytes
and second member the content mime type
Raises:
NotFoundExc if the content is not found
"""
content_data = service.lookup_content(query_string)
filetype = None
language = None
license = None
# requests to the indexer db may fail so properly handle
# those cases in order to avoid content display errors
try:
filetype = service.lookup_content_filetype(query_string)
language = service.lookup_content_language(query_string)
license = service.lookup_content_license(query_string)
except Exception:
pass
mimetype = 'unknown'
encoding = 'unknown'
if filetype:
mimetype = filetype['mimetype']
encoding = filetype['encoding']
# workaround when encountering corrupted data due to implicit
# conversion from bytea to text in the indexer db (see T818)
# TODO: Remove that code when all data have been correctly converted
if mimetype.startswith('\\'):
filetype = None
content_data['error_code'] = 200
content_data['error_message'] = ''
content_data['error_description'] = ''
if not max_size or content_data['length'] < max_size:
try:
content_raw = service.lookup_content_raw(query_string)
except Exception as e:
if raise_if_unavailable:
raise e
else:
content_data['raw_data'] = None
content_data['error_code'] = 404
content_data['error_description'] = \
'The bytes of the content are currently not available in the archive.' # noqa
content_data['error_message'] = \
http_status_code_message[content_data['error_code']]
else:
content_data['raw_data'] = content_raw['data']
if not filetype:
mimetype, encoding = \
get_mimetype_and_encoding_for_content(content_data['raw_data']) # noqa
- # encode textual content to utf-8 if needed
- if reencode and mimetype.startswith('text/'):
- # probably a malformed UTF-8 content, re-encode it
- # by replacing invalid chars with a substitution one
- if encoding == 'unknown-8bit':
- content_data['raw_data'] = \
- content_data['raw_data'].decode('utf-8', 'replace')\
- .encode('utf-8')
- elif 'ascii' not in encoding and encoding not in ['utf-8', 'binary']: # noqa
- content_data['raw_data'] = \
- content_data['raw_data'].decode(encoding, 'replace')\
- .encode('utf-8')
- elif reencode and mimetype.startswith('application/octet-stream'):
- # file may detect a text content as binary
- # so try to decode it for display
- encodings = ['us-ascii']
- encodings += ['iso-8859-%s' % i for i in range(1, 17)]
- for encoding in encodings:
- try:
- content_data['raw_data'] = \
- content_data['raw_data'].decode(encoding)\
- .encode('utf-8')
- except Exception:
- pass
- else:
- # ensure display in content view
- mimetype = 'text/plain'
- break
+ if reencode:
+ mimetype, raw_data = _reencode_content(
+ mimetype, encoding, content_data['raw_data'])
+ content_data['raw_data'] = raw_data
+
else:
content_data['raw_data'] = None
content_data['mimetype'] = mimetype
content_data['encoding'] = encoding
if language:
content_data['language'] = language['lang']
else:
content_data['language'] = 'not detected'
if license:
content_data['licenses'] = ', '.join(license['facts'][0]['licenses'])
else:
content_data['licenses'] = 'not detected'
return content_data
_browsers_supported_image_mimes = set(['image/gif', 'image/png',
'image/jpeg', 'image/bmp',
'image/webp', 'image/svg',
'image/svg+xml'])
def prepare_content_for_display(content_data, mime_type, path):
"""Function that prepares a content for HTML display.
The function tries to associate a programming language to a
content in order to perform syntax highlighting client-side
using highlightjs. The language is determined using either
the content filename or its mime type.
If the mime type corresponds to an image format supported
by web browsers, the content will be encoded in base64
for displaying the image.
Args:
content_data (bytes): raw bytes of the content
mime_type (string): mime type of the content
path (string): path of the content including filename
Returns:
A dict containing the content bytes (possibly different from the one
provided as parameter if it is an image) under the key 'content_data
and the corresponding highlightjs language class under the
key 'language'.
"""
language = highlightjs.get_hljs_language_from_filename(path)
if not language:
language = highlightjs.get_hljs_language_from_mime_type(mime_type)
if not language:
language = 'nohighlight'
elif mime_type.startswith('application/'):
mime_type = mime_type.replace('application/', 'text/')
if mime_type.startswith('image/'):
if mime_type in _browsers_supported_image_mimes:
content_data = base64.b64encode(content_data)
else:
content_data = None
if mime_type.startswith('image/svg'):
mime_type = 'image/svg+xml'
return {'content_data': content_data,
'language': language,
'mimetype': mime_type}
def process_snapshot_branches(snapshot):
"""
Process a dictionary describing snapshot branches: extract those
targeting revisions and releases, put them in two different lists,
then sort those lists in lexicographical order of the branches' names.
Args:
snapshot_branches (dict): A dict describing the branches of a snapshot
as returned for instance by :func:`swh.web.common.service.lookup_snapshot`
Returns:
tuple: A tuple whose first member is the sorted list of branches
targeting revisions and second member the sorted list of branches
targeting releases
""" # noqa
snapshot_branches = snapshot['branches']
branches = {}
branch_aliases = {}
releases = {}
revision_to_branch = defaultdict(set)
revision_to_release = defaultdict(set)
release_to_branch = defaultdict(set)
for branch_name, target in snapshot_branches.items():
if not target:
# FIXME: display branches with an unknown target anyway
continue
target_id = target['target']
target_type = target['target_type']
if target_type == 'revision':
branches[branch_name] = {
'name': branch_name,
'revision': target_id,
}
revision_to_branch[target_id].add(branch_name)
elif target_type == 'release':
release_to_branch[target_id].add(branch_name)
elif target_type == 'alias':
branch_aliases[branch_name] = target_id
# FIXME: handle pointers to other object types
def _enrich_release_branch(branch, release):
releases[branch] = {
'name': release['name'],
'branch_name': branch,
'date': format_utc_iso_date(release['date']),
'id': release['id'],
'message': release['message'],
'target_type': release['target_type'],
'target': release['target'],
}
def _enrich_revision_branch(branch, revision):
branches[branch].update({
'revision': revision['id'],
'directory': revision['directory'],
'date': format_utc_iso_date(revision['date']),
'message': revision['message']
})
releases_info = service.lookup_release_multiple(
release_to_branch.keys()
)
for release in releases_info:
branches_to_update = release_to_branch[release['id']]
for branch in branches_to_update:
_enrich_release_branch(branch, release)
if release['target_type'] == 'revision':
revision_to_release[release['target']].update(
branches_to_update
)
revisions = service.lookup_revision_multiple(
set(revision_to_branch.keys()) | set(revision_to_release.keys())
)
for revision in revisions:
if not revision:
continue
for branch in revision_to_branch[revision['id']]:
_enrich_revision_branch(branch, revision)
for release in revision_to_release[revision['id']]:
releases[release]['directory'] = revision['directory']
for branch_alias, branch_target in branch_aliases.items():
if branch_target in branches:
branches[branch_alias] = dict(branches[branch_target])
else:
snp = service.lookup_snapshot(snapshot['id'],
branches_from=branch_target,
branches_count=1)
if snp and branch_target in snp['branches']:
target_type = snp['branches'][branch_target]['target_type']
target = snp['branches'][branch_target]['target']
if target_type == 'revision':
branches[branch_alias] = snp['branches'][branch_target]
revision = service.lookup_revision(target)
_enrich_revision_branch(branch_alias, revision)
elif target_type == 'release':
release = service.lookup_release(target)
_enrich_release_branch(branch_alias, release)
if branch_alias in branches:
branches[branch_alias]['name'] = branch_alias
ret_branches = list(sorted(branches.values(), key=lambda b: b['name']))
ret_releases = list(sorted(releases.values(), key=lambda b: b['name']))
return ret_branches, ret_releases
def get_snapshot_content(snapshot_id):
"""Returns the lists of branches and releases
associated to a swh snapshot.
That list is put in cache in order to speedup the navigation
in the swh-web/browse ui.
.. warning:: At most 1000 branches contained in the snapshot
will be returned for performance reasons.
Args:
snapshot_id (str): hexadecimal representation of the snapshot
identifier
Returns:
A tuple with two members. The first one is a list of dict describing
the snapshot branches. The second one is a list of dict describing the
snapshot releases.
Raises:
NotFoundExc if the snapshot does not exist
"""
cache_entry_id = 'swh_snapshot_%s' % snapshot_id
cache_entry = cache.get(cache_entry_id)
if cache_entry:
return cache_entry['branches'], cache_entry['releases']
branches = []
releases = []
if snapshot_id:
snapshot = service.lookup_snapshot(
snapshot_id, branches_count=snapshot_content_max_size)
branches, releases = process_snapshot_branches(snapshot)
cache.set(cache_entry_id, {
'branches': branches,
'releases': releases,
})
return branches, releases
def get_origin_visit_snapshot(origin_info, visit_ts=None, visit_id=None,
snapshot_id=None):
"""Returns the lists of branches and releases
associated to a swh origin for a given visit.
The visit is expressed by a timestamp. In the latter case,
the closest visit from the provided timestamp will be used.
If no visit parameter is provided, it returns the list of branches
found for the latest visit.
That list is put in cache in order to speedup the navigation
in the swh-web/browse ui.
.. warning:: At most 1000 branches contained in the snapshot
will be returned for performance reasons.
Args:
origin_info (dict): a dict filled with origin information
(id, url, type)
visit_ts (int or str): an ISO date string or Unix timestamp to parse
visit_id (int): optional visit id for disambiguation in case
several visits have the same timestamp
Returns:
A tuple with two members. The first one is a list of dict describing
the origin branches for the given visit.
The second one is a list of dict describing the origin releases
for the given visit.
Raises:
NotFoundExc if the origin or its visit are not found
"""
visit_info = get_origin_visit(origin_info, visit_ts, visit_id, snapshot_id)
return get_snapshot_content(visit_info['snapshot'])
def gen_link(url, link_text=None, link_attrs={}):
"""
Utility function for generating an HTML link to insert
in Django templates.
Args:
url (str): an url
link_text (str): optional text for the produced link,
if not provided the url will be used
link_attrs (dict): optional attributes (e.g. class)
to add to the link
Returns:
An HTML link in the form '<a href="url">link_text</a>'
"""
attrs = ' '
for k, v in link_attrs.items():
attrs += '%s="%s" ' % (k, v)
if not link_text:
link_text = url
link = '<a%shref="%s">%s</a>' % (attrs, url, link_text)
return mark_safe(link)
def gen_person_link(person_id, person_name, snapshot_context=None,
link_attrs={}):
"""
Utility function for generating a link to a person HTML view
to insert in Django templates.
Args:
person_id (int): a person id
person_name (str): the associated person name
link_attrs (dict): optional attributes (e.g. class)
to add to the link
Returns:
An HTML link in the form '<a href="person_view_url">person_name</a>'
"""
query_params = None
if snapshot_context and snapshot_context['origin_info']:
origin_info = snapshot_context['origin_info']
query_params = {'origin_type': origin_info['type'],
'origin': origin_info['url']}
if 'timestamp' in snapshot_context['url_args']:
query_params['timestamp'] = \
snapshot_context['url_args']['timestamp']
if 'visit_id' in snapshot_context['query_params']:
query_params['visit_id'] = \
snapshot_context['query_params']['visit_id']
elif snapshot_context:
query_params = {'snapshot_id': snapshot_context['snapshot_id']}
person_url = reverse('browse-person', url_args={'person_id': person_id},
query_params=query_params)
return gen_link(person_url, person_name or 'None', link_attrs)
def gen_revision_url(revision_id, snapshot_context=None):
"""
Utility function for generating an url to a revision.
Args:
revision_id (str): a revision id
snapshot_context (dict): if provided, generate snapshot-dependent
browsing url
Returns:
str: The url to browse the revision
"""
query_params = None
if snapshot_context and snapshot_context['origin_info']:
origin_info = snapshot_context['origin_info']
origin_type = snapshot_context['origin_type']
query_params = {'origin_type': origin_type,
'origin': origin_info['url']}
if 'timestamp' in snapshot_context['url_args']:
query_params['timestamp'] = \
snapshot_context['url_args']['timestamp']
if 'visit_id' in snapshot_context['query_params']:
query_params['visit_id'] = \
snapshot_context['query_params']['visit_id']
elif snapshot_context:
query_params = {'snapshot_id': snapshot_context['snapshot_id']}
return reverse('browse-revision',
url_args={'sha1_git': revision_id},
query_params=query_params)
def gen_revision_link(revision_id, shorten_id=False, snapshot_context=None,
link_text=None, link_attrs={}):
"""
Utility function for generating a link to a revision HTML view
to insert in Django templates.
Args:
revision_id (str): a revision id
shorten_id (boolean): whether to shorten the revision id to 7
characters for the link text
snapshot_context (dict): if provided, generate snapshot-dependent
browsing link
link_attrs (dict): optional attributes (e.g. class)
to add to the link
Returns:
str: An HTML link in the form '<a href="revision_url">revision_id</a>'
"""
if not revision_id:
return None
revision_url = gen_revision_url(revision_id, snapshot_context)
if shorten_id:
return gen_link(revision_url, revision_id[:7], link_attrs)
else:
if not link_text:
link_text = revision_id
return gen_link(revision_url, link_text, link_attrs)
def gen_origin_link(origin_info, link_attrs={}):
"""
Utility function for generating a link to a software origin HTML view
to insert in Django templates.
Args:
origin_info (dict): a dict filled with origin information
(id, type, url)
link_attrs (dict): optional attributes (e.g. class)
to add to the link
Returns:
An HTML link in the form '<a href="origin_view_url">Origin: origin_url</a>'
""" # noqa
origin_browse_url = reverse('browse-origin',
url_args={'origin_type': origin_info['type'],
'origin_url': origin_info['url']})
return gen_link(origin_browse_url,
'Origin: ' + origin_info['url'], link_attrs)
def gen_directory_link(sha1_git, link_text=None, link_attrs={}):
"""
Utility function for generating a link to a directory HTML view
to insert in Django templates.
Args:
sha1_git (str): directory identifier
link_text (str): optional text for the generated link
(the generated url will be used by default)
link_attrs (dict): optional attributes (e.g. class)
to add to the link
Returns:
An HTML link in the form '<a href="directory_view_url">link_text</a>'
"""
if not sha1_git:
return None
directory_url = reverse('browse-directory',
url_args={'sha1_git': sha1_git})
if not link_text:
link_text = directory_url
return gen_link(directory_url, link_text, link_attrs)
def gen_snapshot_link(snapshot_id, link_text=None, link_attrs={}):
"""
Utility function for generating a link to a snapshot HTML view
to insert in Django templates.
Args:
snapshot_id (str): snapshot identifier
link_text (str): optional text for the generated link
(the generated url will be used by default)
link_attrs (dict): optional attributes (e.g. class)
to add to the link
Returns:
An HTML link in the form '<a href="snapshot_view_url">link_text</a>'
"""
snapshot_url = reverse('browse-snapshot',
url_args={'snapshot_id': snapshot_id})
if not link_text:
link_text = snapshot_url
return gen_link(snapshot_url, link_text, link_attrs)
def gen_snapshot_directory_link(snapshot_context, revision_id=None,
link_text=None, link_attrs={}):
"""
Utility function for generating a link to a directory HTML view
in the context of a snapshot to insert in Django templates.
Args:
snapshot_context (dict): the snapshot information
revision_id (str): optional revision identifier in order
to use the associated directory
link_text (str): optional text to use for the generated link
link_attrs (dict): optional attributes (e.g. class)
to add to the link
Returns:
An HTML link in the form
'<a href="origin_directory_view_url">origin_directory_view_url</a>'
"""
query_params = {'revision': revision_id}
if snapshot_context['origin_info']:
origin_info = snapshot_context['origin_info']
url_args = {'origin_url': origin_info['url']}
if 'timestamp' in snapshot_context['url_args']:
url_args['timestamp'] = \
snapshot_context['url_args']['timestamp']
if 'visit_id' in snapshot_context['query_params']:
query_params['visit_id'] = \
snapshot_context['query_params']['visit_id']
directory_url = reverse('browse-origin-directory',
url_args=url_args,
query_params=query_params)
else:
url_args = {'snapshot_id': snapshot_context['snapshot_id']}
directory_url = reverse('browse-snapshot-directory',
url_args=url_args,
query_params=query_params)
if not link_text:
link_text = directory_url
return gen_link(directory_url, link_text, link_attrs)
def gen_content_link(sha1_git, link_text=None, link_attrs={}):
"""
Utility function for generating a link to a content HTML view
to insert in Django templates.
Args:
sha1_git (str): content identifier
link_text (str): optional text for the generated link
(the generated url will be used by default)
link_attrs (dict): optional attributes (e.g. class)
to add to the link
Returns:
An HTML link in the form '<a href="content_view_url">link_text</a>'
"""
if not sha1_git:
return None
content_url = reverse('browse-content',
url_args={'query_string': 'sha1_git:' + sha1_git})
if not link_text:
link_text = content_url
return gen_link(content_url, link_text, link_attrs)
def get_revision_log_url(revision_id, snapshot_context=None):
"""
Utility function for getting the URL for a revision log HTML view
(possibly in the context of an origin).
Args:
revision_id (str): revision identifier the history heads to
snapshot_context (dict): if provided, generate snapshot-dependent
browsing link
Returns:
The revision log view URL
"""
query_params = {'revision': revision_id}
if snapshot_context and snapshot_context['origin_info']:
origin_info = snapshot_context['origin_info']
url_args = {'origin_url': origin_info['url']}
if 'timestamp' in snapshot_context['url_args']:
url_args['timestamp'] = \
snapshot_context['url_args']['timestamp']
if 'visit_id' in snapshot_context['query_params']:
query_params['visit_id'] = \
snapshot_context['query_params']['visit_id']
revision_log_url = reverse('browse-origin-log',
url_args=url_args,
query_params=query_params)
elif snapshot_context:
url_args = {'snapshot_id': snapshot_context['snapshot_id']}
revision_log_url = reverse('browse-snapshot-log',
url_args=url_args,
query_params=query_params)
else:
revision_log_url = reverse('browse-revision-log',
url_args={'sha1_git': revision_id})
return revision_log_url
def gen_revision_log_link(revision_id, snapshot_context=None, link_text=None,
link_attrs={}):
"""
Utility function for generating a link to a revision log HTML view
(possibly in the context of an origin) to insert in Django templates.
Args:
revision_id (str): revision identifier the history heads to
snapshot_context (dict): if provided, generate snapshot-dependent
browsing link
link_text (str): optional text to use for the generated link
link_attrs (dict): optional attributes (e.g. class)
to add to the link
Returns:
An HTML link in the form
'<a href="revision_log_view_url">link_text</a>'
"""
if not revision_id:
return None
revision_log_url = get_revision_log_url(revision_id, snapshot_context)
if not link_text:
link_text = revision_log_url
return gen_link(revision_log_url, link_text, link_attrs)
def format_log_entries(revision_log, per_page, snapshot_context=None):
"""
Utility functions that process raw revision log data for HTML display.
Its purpose is to:
* add links to relevant browse views
* format date in human readable format
* truncate the message log
Args:
revision_log (list): raw revision log as returned by the swh-web api
per_page (int): number of log entries per page
snapshot_context (dict): if provided, generate snapshot-dependent
browsing link
"""
revision_log_data = []
for i, rev in enumerate(revision_log):
if i == per_page:
break
author_name = 'None'
author_fullname = 'None'
committer_fullname = 'None'
if rev['author']:
author_name = rev['author']['name'] or rev['author']['fullname']
author_fullname = rev['author']['fullname']
if rev['committer']:
committer_fullname = rev['committer']['fullname']
author_date = format_utc_iso_date(rev['date'])
committer_date = format_utc_iso_date(rev['committer_date'])
tooltip = 'revision %s\n' % rev['id']
tooltip += 'author: %s\n' % author_fullname
tooltip += 'author date: %s\n' % author_date
tooltip += 'committer: %s\n' % committer_fullname
tooltip += 'committer date: %s\n\n' % committer_date
if rev['message']:
tooltip += textwrap.indent(rev['message'], ' '*4)
revision_log_data.append({
'author': author_name,
'id': rev['id'][:7],
'message': rev['message'],
'date': author_date,
'commit_date': committer_date,
'url': gen_revision_url(rev['id'], snapshot_context),
'tooltip': tooltip
})
return revision_log_data
# list of origin types that can be found in the swh archive
# TODO: retrieve it dynamically in an efficient way instead
# of hardcoding it
_swh_origin_types = ['git', 'svn', 'deb', 'hg', 'ftp', 'deposit', 'pypi']
def get_origin_info(origin_url, origin_type=None):
"""
Get info about a software origin.
Its main purpose is to automatically find an origin type
when it is not provided as parameter.
Args:
origin_url (str): complete url of a software origin
origin_type (str): optional origin type
Returns:
A dict with the following entries:
* type: the origin type
* url: the origin url
* id: the internal id of the origin
"""
if origin_type:
return service.lookup_origin({'type': origin_type,
'url': origin_url})
else:
for origin_type in _swh_origin_types:
try:
origin_info = service.lookup_origin({'type': origin_type,
'url': origin_url})
return origin_info
except Exception:
pass
raise NotFoundExc('Origin with url %s not found!' % origin_url)
def get_snapshot_context(snapshot_id=None, origin_type=None, origin_url=None,
timestamp=None, visit_id=None):
"""
Utility function to compute relevant information when navigating
the archive in a snapshot context. The snapshot is either
referenced by its id or it will be retrieved from an origin visit.
Args:
snapshot_id (str): hexadecimal representation of a snapshot identifier,
all other parameters will be ignored if it is provided
origin_type (str): the origin type (git, svn, deposit, ...)
origin_url (str): the origin_url (e.g. https://github.com/(user)/(repo)/)
timestamp (str): a datetime string for retrieving the closest
visit of the origin
visit_id (int): optional visit id for disambiguation in case
of several visits with the same timestamp
Returns:
A dict with the following entries:
* origin_info: dict containing origin information
* visit_info: dict containing visit information
* branches: the list of branches for the origin found
during the visit
* releases: the list of releases for the origin found
during the visit
* origin_browse_url: the url to browse the origin
* origin_branches_url: the url to browse the origin branches
* origin_releases_url': the url to browse the origin releases
* origin_visit_url: the url to browse the snapshot of the origin
found during the visit
* url_args: dict containing url arguments to use when browsing in
the context of the origin and its visit
Raises:
NotFoundExc: if no snapshot is found for the visit of an origin.
""" # noqa
origin_info = None
visit_info = None
url_args = None
query_params = {}
branches = []
releases = []
browse_url = None
visit_url = None
branches_url = None
releases_url = None
swh_type = 'snapshot'
if origin_url:
swh_type = 'origin'
origin_info = get_origin_info(origin_url, origin_type)
visit_info = get_origin_visit(origin_info, timestamp, visit_id,
snapshot_id)
fmt_date = format_utc_iso_date(visit_info['date'])
visit_info['fmt_date'] = fmt_date
snapshot_id = visit_info['snapshot']
if not snapshot_id:
raise NotFoundExc('No snapshot associated to the visit of origin '
'%s on %s' % (origin_url, fmt_date))
# provided timestamp is not necessarily equals to the one
# of the retrieved visit, so get the exact one in order
# use it in the urls generated below
if timestamp:
timestamp = visit_info['date']
branches, releases = \
get_origin_visit_snapshot(origin_info, timestamp, visit_id,
snapshot_id)
url_args = {'origin_type': origin_type,
'origin_url': origin_info['url']}
query_params = {'visit_id': visit_id}
browse_url = reverse('browse-origin-visits',
url_args=url_args)
if timestamp:
url_args['timestamp'] = format_utc_iso_date(timestamp,
'%Y-%m-%dT%H:%M:%S')
visit_url = reverse('browse-origin-directory',
url_args=url_args,
query_params=query_params)
visit_info['url'] = visit_url
branches_url = reverse('browse-origin-branches',
url_args=url_args,
query_params=query_params)
releases_url = reverse('browse-origin-releases',
url_args=url_args,
query_params=query_params)
elif snapshot_id:
branches, releases = get_snapshot_content(snapshot_id)
url_args = {'snapshot_id': snapshot_id}
browse_url = reverse('browse-snapshot',
url_args=url_args)
branches_url = reverse('browse-snapshot-branches',
url_args=url_args)
releases_url = reverse('browse-snapshot-releases',
url_args=url_args)
releases = list(reversed(releases))
snapshot_size = service.lookup_snapshot_size(snapshot_id)
is_empty = sum(snapshot_size.values()) == 0
swh_snp_id = persistent_identifier('snapshot', snapshot_id)
return {
'swh_type': swh_type,
'swh_object_id': swh_snp_id,
'snapshot_id': snapshot_id,
'snapshot_size': snapshot_size,
'is_empty': is_empty,
'origin_info': origin_info,
# keep track if the origin type was provided as url argument
'origin_type': origin_type,
'visit_info': visit_info,
'branches': branches,
'releases': releases,
'branch': None,
'release': None,
'browse_url': browse_url,
'branches_url': branches_url,
'releases_url': releases_url,
'url_args': url_args,
'query_params': query_params
}
# list of common readme names ordered by preference
# (lower indices have higher priority)
_common_readme_names = [
"readme.markdown",
"readme.md",
"readme.rst",
"readme.txt",
"readme"
]
def get_readme_to_display(readmes):
"""
Process a list of readme files found in a directory
in order to find the adequate one to display.
Args:
readmes: a list of dict where keys are readme file names and values
are readme sha1s
Returns:
A tuple (readme_name, readme_sha1)
"""
readme_name = None
readme_url = None
readme_sha1 = None
readme_html = None
lc_readmes = {k.lower(): {'orig_name': k, 'sha1': v}
for k, v in readmes.items()}
# look for readme names according to the preference order
# defined by the _common_readme_names list
for common_readme_name in _common_readme_names:
if common_readme_name in lc_readmes:
readme_name = lc_readmes[common_readme_name]['orig_name']
readme_sha1 = lc_readmes[common_readme_name]['sha1']
readme_url = reverse('browse-content-raw',
url_args={'query_string': readme_sha1},
query_params={'reencode': 'true'})
break
# otherwise pick the first readme like file if any
if not readme_name and len(readmes.items()) > 0:
readme_name = next(iter(readmes))
readme_sha1 = readmes[readme_name]
readme_url = reverse('browse-content-raw',
url_args={'query_string': readme_sha1},
query_params={'reencode': 'true'})
# convert rst README to html server side as there is
# no viable solution to perform that task client side
if readme_name and readme_name.endswith('.rst'):
cache_entry_id = 'readme_%s' % readme_sha1
cache_entry = cache.get(cache_entry_id)
if cache_entry:
readme_html = cache_entry
else:
try:
rst_doc = request_content(readme_sha1)
readme_html = pypandoc.convert_text(rst_doc['raw_data'],
'html', format='rst')
cache.set(cache_entry_id, readme_html)
except Exception:
readme_html = 'Readme bytes are not available'
return readme_name, readme_url, readme_html
def get_swh_persistent_ids(swh_objects, snapshot_context=None):
"""
Returns a list of dict containing info related to persistent
identifiers of swh objects.
Args:
swh_objects (list): a list of dict with the following keys:
* type: swh object type (content/directory/release/revision/snapshot)
* id: swh object id
snapshot_context (dict): optional parameter describing the snapshot in which
the object has been found
Returns:
list: a list of dict with the following keys:
* object_type: the swh object type (content/directory/release/revision/snapshot)
* object_icon: the swh object icon to use in HTML views
* swh_id: the computed swh object persistent identifier
* swh_id_url: the url resolving the persistent identifier
* show_options: boolean indicating if the persistent id options must
be displayed in persistent ids HTML view
""" # noqa
swh_ids = []
for swh_object in swh_objects:
if not swh_object['id']:
continue
swh_id = get_swh_persistent_id(swh_object['type'], swh_object['id'])
show_options = swh_object['type'] == 'content' or \
(snapshot_context and snapshot_context['origin_info'] is not None)
object_icon = swh_object_icons[swh_object['type']]
swh_ids.append({
'object_type': swh_object['type'],
'object_icon': object_icon,
'swh_id': swh_id,
'swh_id_url': reverse('browse-swh-id',
url_args={'swh_id': swh_id}),
'show_options': show_options
})
return swh_ids
diff --git a/swh/web/tests/browse/views/data/content_test_data.py b/swh/web/tests/browse/views/data/content_test_data.py
index 3250a594..75a6fc75 100644
--- a/swh/web/tests/browse/views/data/content_test_data.py
+++ b/swh/web/tests/browse/views/data/content_test_data.py
@@ -1,232 +1,83 @@
-# Copyright (C) 2017-2018 The Software Heritage developers
+# Copyright (C) 2017-2019 The Software Heritage developers
# See the AUTHORS file at the top-level directory of this distribution
# License: GNU Affero General Public License version 3, or any later version
# See top-level LICENSE file for more information
# flake8: noqa
import os
stub_content_root_dir = '08e8329257dad3a3ef7adea48aa6e576cd82de5b'
stub_content_text_file = \
"""
/* This file is part of the KDE project
*
* This library is free software; you can redistribute it and/or
* modify it under the terms of the GNU Library General Public
* License as published by the Free Software Foundation; either
* version 2 of the License, or (at your option) any later version.
*
* This library is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
* Library General Public License for more details.
*
* You should have received a copy of the GNU Library General Public License
* along with this library; see the file COPYING.LIB. If not, write to
* the Free Software Foundation, Inc., 51 Franklin Street, Fifth Floor,
* Boston, MA 02110-1301, USA.
*/
#ifndef KATE_SESSION_TEST_H
#define KATE_SESSION_TEST_H
#include <QObject>
class KateSessionTest : public QObject
{
Q_OBJECT
private Q_SLOTS:
void init();
void cleanup();
void initTestCase();
void cleanupTestCase();
void create();
void createAnonymous();
void createAnonymousFrom();
void createFrom();
void documents();
void setFile();
void setName();
void timestamp();
private:
class QTemporaryFile *m_tmpfile;
};
#endif
"""
-stub_content_text_file_no_highlight = \
-"""
- GNU GENERAL PUBLIC LICENSE
- Version 3, 29 June 2007
-
- Copyright (C) 2007 Free Software Foundation, Inc. <http://fsf.org/>
- Everyone is permitted to copy and distribute verbatim copies
- of this license document, but changing it is not allowed.
-
- Preamble
-
- The GNU General Public License is a free, copyleft license for
-software and other kinds of works.
-
- The licenses for most software and other practical works are designed
-to take away your freedom to share and change the works. By contrast,
-the GNU General Public License is intended to guarantee your freedom to
-share and change all versions of a program--to make sure it remains free
-software for all its users. We, the Free Software Foundation, use the
-GNU General Public License for most of our software; it applies also to
-any other work released this way by its authors. You can apply it to
-your programs, too.
-
- When we speak of free software, we are referring to freedom, not
-price. Our General Public Licenses are designed to make sure that you
-have the freedom to distribute copies of free software (and charge for
-them if you wish), that you receive source code or can get it if you
-want it, that you can change the software or use pieces of it in new
-free programs, and that you know you can do these things.
-
- To protect your rights, we need to prevent others from denying you
-these rights or asking you to surrender the rights. Therefore, you have
-certain responsibilities if you distribute copies of the software, or if
-you modify it: responsibilities to respect the freedom of others.
-
- For example, if you distribute copies of such a program, whether
-gratis or for a fee, you must pass on to the recipients the same
-freedoms that you received. You must make sure that they, too, receive
-or can get the source code. And you must show them these terms so they
-know their rights.
-
- Developers that use the GNU GPL protect your rights with two steps:
-(1) assert copyright on the software, and (2) offer you this License
-giving you legal permission to copy, distribute and/or modify it.
-
- For the developers' and authors' protection, the GPL clearly explains
-that there is no warranty for this free software. For both users' and
-authors' sake, the GPL requires that modified versions be marked as
-changed, so that their problems will not be attributed erroneously to
-authors of previous versions.
-
- Some devices are designed to deny users access to install or run
-modified versions of the software inside them, although the manufacturer
-can do so. This is fundamentally incompatible with the aim of
-protecting users' freedom to change the software. The systematic
-pattern of such abuse occurs in the area of products for individuals to
-use, which is precisely where it is most unacceptable. Therefore, we
-have designed this version of the GPL to prohibit the practice for those
-products. If such problems arise substantially in other domains, we
-stand ready to extend this provision to those domains in future versions
-of the GPL, as needed to protect the freedom of users.
-
- Finally, every program is threatened constantly by software patents.
-States should not allow patents to restrict development and use of
-software on general-purpose computers, but in those that do, we wish to
-avoid the special danger that patents applied to a free program could
-make it effectively proprietary. To prevent this, the GPL assures that
-patents cannot be used to render the program non-free.
-"""
-
stub_content_text_data = {
'checksums': {
'sha1': '5ecd9f37b7a2d2e9980d201acd6286116f2ba1f1',
'sha1_git': '537b47f68469c1c916c1bfbc072599133bfcbb21',
'sha256': 'b3057544f04e5821ab0e2a007e2ceabd7de2dfb1d42a764f1de8d0d2eff80006',
'blake2s256': '25117fa9f124d5b771a0a7dfca9c7a57247d81f8343334b4b41c782c7f7ed64d'
},
'length': 1317,
'raw_data': str.encode(stub_content_text_file),
'mimetype': 'text/x-c++',
'encoding': 'us-ascii',
'language': 'c++',
'licenses': 'GPL',
'error_code': 200,
'error_message': '',
'error_description': ''
}
-stub_content_text_no_highlight_data = {
- 'checksums': {
- 'sha1': '8624bcdae55baeef00cd11d5dfcfa60f68710a02',
- 'sha1_git': '94a9ed024d3859793618152ea559a168bbcbb5e2',
- 'sha256': '8ceb4b9ee5adedde47b31e975c1d90c73ad27b6b165a1dcd80c7c545eb65b903',
- 'blake2s256': '38702b7168c7785bfe748b51b45d9856070ba90f9dc6d90f2ea75d4356411ffe'
- },
- 'length': 35147,
- 'raw_data': str.encode(stub_content_text_file_no_highlight),
- 'mimetype': 'text/plain',
- 'encoding': 'us-ascii',
- 'language': 'not detected',
- 'licenses': 'GPL',
- 'error_code': 200,
- 'error_message': '',
- 'error_description': ''
-}
-
stub_content_text_path = 'kate/autotests/session_test.h'
-stub_content_text_path_with_root_dir = stub_content_root_dir + '/' + stub_content_text_path
-
-stub_content_bin_filename = 'swh-logo.png'
-
-png_file_path = os.path.dirname(__file__) + '/' + stub_content_bin_filename
-with open(png_file_path, 'rb') as png_file:
- stub_content_bin_data = {
- 'checksums': {
- 'sha1': 'd0cec0fc2d795f0077c18d51578cdb228eaf6a99',
- 'sha1_git': '02328b91cfad800e1d2808cfb379511b79679ebc',
- 'sha256': 'e290592e2cfa9767497011bda4b7e273b4cf29e7695d72ecacbd723008a29144',
- 'blake2s256': '7177cad95407952e362ee326a800a9d215ccd619fdbdb735bb51039be81ab9ce'
- },
- 'length': 18063,
- 'raw_data': png_file.read(),
- 'mimetype': 'image/png',
- 'encoding': 'binary',
- 'language': 'not detected',
- 'licenses': 'not detected',
- 'error_code': 200,
- 'error_message': '',
- 'error_description': ''
- }
-
-_non_utf8_encoding_file_path = os.path.dirname(__file__) + '/iso-8859-1_encoded_content'
-
-non_utf8_encoded_content_data = {
- 'checksums': {
- 'sha1': '62cb71aa3534a03c12572157d20fa893753b03b6',
- 'sha1_git': '2f7470d0b26108130e71087e42a53c032473499c',
- 'sha256': 'aaf364ccd3acb546829ccc0e8e5e293e924c8a2e55a67cb739d249016e0034ed',
- 'blake2s256': 'b7564932460a7c2697c53bd55bd855272490da511d64b20c5a04f636dc9ac467'
- },
- 'length': 111000,
- 'error_code': 200,
- 'error_message': '',
- 'error_description': ''
-}
-
-non_utf8_encoding = 'iso-8859-1'
-
-with open(_non_utf8_encoding_file_path, 'rb') as iso88591_file:
- non_utf8_encoded_content = iso88591_file.read()
-
-stub_content_too_large_data = {
- 'checksums': {
- 'sha1': '8624bcdae55baeef00cd11d5dfcfa60f68710a02',
- 'sha1_git': '94a9ed024d3859793618152ea559a168bbcbb5e2',
- 'sha256': '8ceb4b9ee5adedde47b31e975c1d90c73ad27b6b165a1dcd80c7c545eb65b903',
- 'blake2s256': '38702b7168c7785bfe748b51b45d9856070ba90f9dc6d90f2ea75d4356411ffe'
- },
- 'length': 3000000,
- 'raw_data': None,
- 'mimetype': 'text/plain',
- 'encoding': 'us-ascii',
- 'language': 'not detected',
- 'licenses': 'GPL',
- 'error_code': 200,
- 'error_message': '',
- 'error_description': ''
-}
\ No newline at end of file
diff --git a/swh/web/tests/browse/views/data/iso-8859-1_encoded_content b/swh/web/tests/browse/views/data/iso-8859-1_encoded_content
deleted file mode 100644
index 2f7470d0..00000000
--- a/swh/web/tests/browse/views/data/iso-8859-1_encoded_content
+++ /dev/null
@@ -1,279 +0,0 @@
-<?php
-/*****************************************************************************
- * Bechamail script.
- * This program is free software; you can redistribute it and/or
- * modify it under the terms of the GNU General Public License
- * as published by the Free Software Foundation; either version 2
- * of the License, or (at your option) any later version.
- *
- * This program is distributed in the hope that it will be useful,
- * but WITHOUT ANY WARRANTY; without even the implied warranty of
- * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
- * GNU General Public License for more details.
- *
- * You should have received a copy of the GNU General Public License
- * along with this program; if not, write to the Free Software
- * Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA,
- * or go to http://www.gnu.org/copyleft/gpl.html
- *
- * Developer(s): 2004, Robert Sebille - robert.sebille(at)cassiopea(dot)org
- * Company: Cassiopea asbl - Belgique
- *****************************************************************************/
-
-// Placement d'un cookie de v�rification pour le login
// SETCOOKIE("CartableCookieEnabled", "test");
// Maintenant via config.inc
-
-session_start();
-
-//Page accessible � tous
-
/*
-if (!isset($_SESSION['S_id_niveau']) or $_SESSION['S_id_niveau'] < 0 or $_SESSION['S_id_niveau'] > 4)
- {header("Location: login.php?mess=".$mess_inc_non_autorise); exit();}
-
-*/
-
-if (isset($_POST['f_deconnect'])) {
- //log
- require "../utiles/config.inc";
- require "../utiles/fonctions.inc";
- if ($conf_log) {ajout_to_log("LOGOUT");}
- session_destroy(); header("Location: login.php?mess=".urlencode("Vous �tes d�connect�")); exit();
- }
-
-
-// Load TemTab class et messages.inc
-require "../classes/class.TemTab.php";
-require "../classes/class.ReqSql.php";
-require "../utiles/config.inc";
-require "../utiles/c.inc";
-require "../utiles/fonctions.inc";
-
-require "../utiles/secure-session.inc";
-
-// Acc�s ok pour niveaux 1 � 2
-niveau_ok(1,2);
-
-// compute starting time with microseconds precision
-if ($conf_aff_req_sql) {
- $mtime = explode(" ",microtime());
- $starttime = $mtime[1] + $mtime[0];
- }
-
-// Soyons prudent et portable, d�sactivons magic_quote pour ce script. merci nexen ;)
if (get_magic_quotes_gpc()) {
- function stripslashes_deep($value)
- {
- $value = is_array($value) ?
- array_map('stripslashes_deep', $value) :
- stripslashes($value);
-
- return $value;
- }
-
- $_POST = array_map('stripslashes_deep', $_POST);
- $_GET = array_map('stripslashes_deep', $_GET);
- $_COOKIE = array_map('stripslashes_deep', $_COOKIE);
-}
-
-/**************************/
-/* Cr�e les objets TemTab */
-/**************************/
-
- // top
- $template_file = "../html/meta.html";
- $template_meta = new TemTab($template_file);
-
- if ($_SESSION["S_niveau"] == 1) {$template_file = "../html/tete-admin-statistiques.html";}
- if ($_SESSION["S_niveau"] == 2) {$template_file = "../html/tete-statistiques.html";}
- $template_tete = new TemTab($template_file);
-
- // corps
- $template_file = "../html/statistiques.html";
- $template = new TemTab($template_file);
-
- // pied
$template_file = "../html/pied.html";
- $template_pied = new TemTab($template_file);
-
-// Some initializations
-$sel="";$from="";$whe="";$col_val="";$affmess="";
-
-
-/************************/
-/* Traitement du script */
-/************************/
-
- // Pr�paration de l'info alias / destination g�n�rique
- $info_alias_dest_gen = ""; $client_auth_gen = false;
- $client_auth_gen = ($conf_alias_dest_gen or $_SESSION["S_niveau"] == 1);
- if (!$client_auth_gen) {
- $info_alias_dest_gen = "Vous n\'avez pas le droit de modifier des alias ou destinations g�n�riques, du type "@domain.tld " ! Vous devez demander cette modification � l\'administrateur du syst�me.";
- }
- else {
- $info_alias_dest_gen = "Vous avez le droit de modifier des alias ou destinations g�n�riques, du type "@domain.tld ".";
- }
-
- $domaines = str_replace(" ", "", $_SESSION["S_domaines"]);
-
- // construction de la clause where en fonction des domaines autoris�s par le login
- $whe_dom = "";
- $whe_dom = $conf_champ_alias." like \"%".$domaines;
- $whe_dom = str_replace(",", "\" or ".$conf_champ_alias." like \"%", $whe_dom);
- $whe_dom = $whe_dom."\"";
-
- // recherche des alias et destinations g�n�riques
- $alias_dest_gen = new ReqSql($hm_, $um_, $pm_, $nm_);
- $sel = $conf_champ_alias.", ".$conf_champ_destination;
- $from = $conf_table_alias;
- $order = $conf_champ_alias." ASC";
- $alias_dest_gen->RS_select($sel, $from, $whe_dom, $order);
- mysql_free_result($alias_dest_gen->RS_res);
-
- $alias_dest_gen_array = array();
- foreach($alias_dest_gen->RS_ligne as $val1) {
- foreach($val1 as $key2 => $val2) {
- if (($key2 == "0" or $key2 == "1") and (substr($val2, 0, 1) == "@")) {
- array_push($alias_dest_gen_array, "<a href='modif-alias.php?alias=".$val1["0"]."&dest=".$val1["1"]."'>".$val1["0"]."</a> => ".$val1["1"]);
- } // if (($key2 == "0" or $key2 == "1") and ..
- } // foreach($val1 as $key2 => $val2)
- } // foreach($alias_dest_gen->RS_ligne as $val1)
-
- $alias_dest_gen_array = array_unique($alias_dest_gen_array);
-
-
- // Recherche des domaines ayant un alias
- $dom_login = new ReqSql($hm_, $um_, $pm_, $nm_);
-// Pas mal, le select et le group by, l� ;-)
- $sel = "SUBSTRING(".$conf_champ_alias.", LOCATE('@', ".$conf_champ_alias.") + 1, LENGTH(".$conf_champ_alias.")) as nom_domaine, count( SUBSTRING(".$conf_champ_alias.", LOCATE('@', ".$conf_champ_alias.") + 1, LENGTH(".$conf_champ_alias."))) as nb_domaine";
- $from = $conf_table_alias;
- $group = "SUBSTRING(".$conf_champ_alias.", LOCATE('@', ".$conf_champ_alias.") + 1, LENGTH(".$conf_champ_alias."))";
- $dom_login->RS_select_group($sel, $from, $whe_dom, $group);
- mysql_free_result($dom_login->RS_res);
-
- // Ajout des domaines du fichier de config n'ayant pas d'alias, puis tri
- $domaine_array = array();$dom_array_compare = array();$dom_array_all = array();
- // d�doublonner le tableau de la requete + preparer un tableau simple de comparaison
- foreach($dom_login->RS_ligne as $val) {
- array_push($domaine_array, array_unique($val));
- array_push($dom_array_compare, $val["0"]);
- }
-
- // Tableau simple de tous les domaines autoris�
- $dom_array_all = explode(",", $domaines);
-
- // ajout des domaines du fichier de config n'ayant pas d'alias a ceux qui en ont
- foreach($dom_array_all as $val) {
- if (!in_array($val, $dom_array_compare)) {array_push($domaine_array, array($val, 0));}
- }
-
- // Tri (voir bas de page : http://www.nexen.net/docs/php/annotee/function.array-multisort.php )
- foreach ($domaine_array as $key => $row) {
- $ligne0[$key] = $row['0'];
- $ligne1[$key] = $row['1'];
- }
- array_multisort($ligne0, SORT_ASC, $ligne1, SORT_DESC, $domaine_array);
-
- // Correspondance login / domaines et liste des domaines de la table transport
- $f_base_login = "";$f_base_domaines = "";
- if ($_SESSION["S_niveau"] == 1) {
- $liste_login = new ReqSql($hl_, $ul_, $pl_, $nl_);
- $sel = $conf_champ_login.",".$conf_champ_domaines;
- $from = $conf_table_login;
- $liste_login->RS_select($sel, $from, "", $conf_champ_login);
- mysql_free_result($liste_login->RS_res);
- $f_base_login = "Login => domaines g�r�s<br /><span class=\"note\">(".$liste_login->RS_nb_ligne." login trouv�(s) dans dans la table <i>".$conf_table_login."</i> de la base de donn�es <i>".$conf_base_login."</i>. All = tous les domaines [super admin] )</span>";
-
- $liste_dom = new ReqSql($hm_, $um_, $pm_, $nm_);
- $sel = $conf_champ_domaine.",".$conf_champ_transport;
- $from = $conf_table_transport;
- $liste_dom->RS_select($sel, $from, "", $conf_champ_domaine);
- mysql_free_result($liste_dom->RS_res);
- $f_base_domaines = "Domaines [<i>transport</i>]<br /><span class=\"note\">(".$liste_dom->RS_nb_ligne." domaine(s) trouv�(s) dans la table <i>".$conf_table_transport."</i> de la base de donn�es <i>".$conf_base_mail."</i>.</span>";
-
- } // if ($_SESSION["S_niveau"] == 1)
-
-
-/***********************************/
-/* traitement des variables TemTab */
-/***********************************/
-
- // Affichage des balises meta
- $template_meta->replace_var("t_meta_charset", $conf_meta_charset);
- $template_meta->replace_var("t_meta_general_keywords", $conf_meta_general_keywords);
- $template_meta->replace_var("t_meta_particulier_keywords", $meta_particulier_keywords);
- $template_meta->replace_var("t_meta_copyright", $conf_meta_copyright);
- $template_meta->replace_var("t_meta_author", $conf_meta_author);
- $template_meta->replace_var("t_meta_generator", $conf_meta_generator);
- $template_meta->replace_var("t_meta_description", $conf_meta_description);
-
-
- // Affichage des titres
- $template_meta->replace_var("t_nom_site", $conf_nom_site);
-
- // Affichage de l(identifiant du site
- $template_tete->replace_var("t_conf_name_log", $conf_name_log);
-
- // formulaire d�connexion
- $template_tete->replace_var("t_script_name", $conf_script_name);
-
- // Affichage des alias et destination g�n�riques
- $template->replace_loop_vars_one("loop_gen", $alias_dest_gen_array);
-
- // Affichage des statistiques des alias
- $template->replace_loop_vars("loop_stats", $domaine_array);
-
- // Affichage des statistiques des comptes
- $template->replace_loop_vars("loop_stats_comptes", 0);
-
- // Affichage des correspondances login / domaines
- $template->replace_var("t_base_login_domaines", $f_base_login);
- if ($_SESSION["S_niveau"] == 1) {$template->replace_loop_vars("loop_liste_login_domaines", $liste_login->RS_ligne);}
- else {@$template->replace_loop_vars("loop_liste_login_domaines", $login_domaines_array);} // tableau vide si niveau 2
-
- // Affichage des domaines de la table transport
- $template->replace_var("t_base_domaines", $f_base_domaines);
- if ($_SESSION["S_niveau"] == 1) {$template->replace_loop_vars("loop_liste_domaines", $liste_dom->RS_ligne);}
- else {@$template->replace_loop_vars("loop_liste_domaines", $login_domaines_array);} // tableau vide si niveau 2
-
- // Affichage de l'info alias / destination g�n�rique
- $template->replace_var("t_info_alias_dest_gen", $info_alias_dest_gen);
-
- // affichage du login
- $f_login = $_SESSION["S_login"];
- $template->replace_var("t_login", $f_login);
-
-
-// pr�pare un message �ventuellement pass� par GET (d�connexion, en l'occurence)
-if (isset($_GET["mess"])) {$affmess = stripslashes($_GET["mess"]);}
-if (!isset($affmess)) {$affmess = "";}
-if (isset($_GET["mess"]) or $affmess != "") {
- $template->replace_var("t_mess_accueil", "<div class=\"mess\">".$affmess."</div><div> </div>");
- }
-
-
-/*****************************************/
-/* Termine la construction des templates */
-/*****************************************/
-
- // get rid of unused variables
- $template_meta->remove_all_vars();
- $template_tete->remove_all_vars();
- $template->remove_all_vars();
- $template_pied->remove_all_vars();
-
- // and finally get the outputs
- print ($template_meta->get_output());
- print ($template_tete->get_output());
- print ($template->get_output());
- print ($template_pied->get_output());
-
-
-// compute elapsed time with microseconds precision
-if ($conf_aff_req_sql) {
- $mtime = explode(" ",microtime());
- $endtime = $mtime[1] + $mtime[0] - $starttime;
- print "\n";
- print "<div><div class=\"pied\">Dur�e du traitement: ".$endtime." microsecondes</div></div>\n<br>";
- }
-
-
-
-?>
-
diff --git a/swh/web/tests/browse/views/data/swh-logo.png b/swh/web/tests/browse/views/data/swh-logo.png
deleted file mode 100644
index 3cbb7196..00000000
Binary files a/swh/web/tests/browse/views/data/swh-logo.png and /dev/null differ
diff --git a/swh/web/tests/browse/views/test_content.py b/swh/web/tests/browse/views/test_content.py
index 1a45cf6f..f88bf07a 100644
--- a/swh/web/tests/browse/views/test_content.py
+++ b/swh/web/tests/browse/views/test_content.py
@@ -1,326 +1,350 @@
-# Copyright (C) 2017-2018 The Software Heritage developers
+# Copyright (C) 2017-2019 The Software Heritage developers
# See the AUTHORS file at the top-level directory of this distribution
# License: GNU Affero General Public License version 3, or any later version
# See top-level LICENSE file for more information
-import base64
-
from unittest.mock import patch
from django.utils.html import escape
-from django.utils.encoding import DjangoUnicodeDecodeError
-from swh.web.browse.utils import get_mimetype_and_encoding_for_content
+from hypothesis import given
+
+from swh.web.browse.utils import (
+ get_mimetype_and_encoding_for_content, prepare_content_for_display,
+ _reencode_content
+)
from swh.web.common.exc import NotFoundExc
from swh.web.common.utils import reverse, get_swh_persistent_id
from swh.web.common.utils import gen_path_info
-from swh.web.tests.testcase import WebTestCase
-
-from .data.content_test_data import (
- stub_content_text_data,
- stub_content_text_path_with_root_dir,
- stub_content_bin_data,
- stub_content_bin_filename,
- stub_content_text_no_highlight_data,
- non_utf8_encoded_content_data,
- non_utf8_encoded_content,
- non_utf8_encoding,
- stub_content_too_large_data
+from swh.web.tests.strategies import (
+ content, content_text_non_utf8, content_text_no_highlight,
+ content_image_type, content_text, invalid_sha1, unknown_content
)
+from swh.web.tests.testcase import WebTestCase
class SwhBrowseContentTest(WebTestCase):
- @patch('swh.web.browse.views.content.request_content')
- def test_content_view_text(self, mock_request_content):
- mock_request_content.return_value = stub_content_text_data
+ @given(content())
+ def test_content_view_text(self, content):
- sha1_git = stub_content_text_data['checksums']['sha1_git']
+ sha1_git = content['sha1_git']
url = reverse('browse-content',
- url_args={'query_string': stub_content_text_data['checksums']['sha1']}) # noqa
+ url_args={'query_string': content['sha1']},
+ query_params={'path': content['path']})
url_raw = reverse('browse-content-raw',
- url_args={'query_string': stub_content_text_data['checksums']['sha1']}) # noqa
+ url_args={'query_string': content['sha1']})
resp = self.client.get(url)
+ content_display = self._process_content_for_display(content)
+ mimetype = content_display['mimetype']
+
self.assertEqual(resp.status_code, 200)
self.assertTemplateUsed('browse/content.html')
- self.assertContains(resp, '<code class="cpp">')
- self.assertContains(resp, escape(stub_content_text_data['raw_data']))
+ if mimetype.startswith('text/'):
+ self.assertContains(resp, '<code class="%s">' %
+ content_display['language'])
+ self.assertContains(resp, escape(content_display['content_data']))
self.assertContains(resp, url_raw)
swh_cnt_id = get_swh_persistent_id('content', sha1_git)
swh_cnt_id_url = reverse('browse-swh-id',
url_args={'swh_id': swh_cnt_id})
self.assertContains(resp, swh_cnt_id)
self.assertContains(resp, swh_cnt_id_url)
- @patch('swh.web.browse.views.content.request_content')
- def test_content_view_text_no_highlight(self, mock_request_content):
- mock_request_content.return_value = stub_content_text_no_highlight_data
+ @given(content_text_no_highlight())
+ def test_content_view_text_no_highlight(self, content):
- sha1_git = stub_content_text_no_highlight_data['checksums']['sha1_git']
+ sha1_git = content['sha1_git']
url = reverse('browse-content',
- url_args={'query_string': stub_content_text_no_highlight_data['checksums']['sha1']}) # noqa
+ url_args={'query_string': content['sha1']})
url_raw = reverse('browse-content-raw',
- url_args={'query_string': stub_content_text_no_highlight_data['checksums']['sha1']}) # noqa
+ url_args={'query_string': content['sha1']})
resp = self.client.get(url)
+ content_display = self._process_content_for_display(content)
+
self.assertEqual(resp.status_code, 200)
self.assertTemplateUsed('browse/content.html')
self.assertContains(resp, '<code class="nohighlight">')
- self.assertContains(resp, escape(stub_content_text_no_highlight_data['raw_data'])) # noqa
+ self.assertContains(resp, escape(content_display['content_data'])) # noqa
self.assertContains(resp, url_raw)
swh_cnt_id = get_swh_persistent_id('content', sha1_git)
swh_cnt_id_url = reverse('browse-swh-id',
url_args={'swh_id': swh_cnt_id})
self.assertContains(resp, swh_cnt_id)
self.assertContains(resp, swh_cnt_id_url)
- @patch('swh.web.browse.utils.service')
- def test_content_view_no_utf8_text(self, mock_service):
- mock_service.lookup_content.return_value = \
- non_utf8_encoded_content_data
+ @given(content_text_non_utf8())
+ def test_content_view_no_utf8_text(self, content):
- mock_service.lookup_content_raw.return_value = \
- {'data': non_utf8_encoded_content}
+ sha1_git = content['sha1_git']
- mock_service.lookup_content_filetype.return_value = None
- mock_service.lookup_content_language.return_value = None
- mock_service.lookup_content_license.return_value = None
+ url = reverse('browse-content',
+ url_args={'query_string': content['sha1']})
- sha1_git = non_utf8_encoded_content_data['checksums']['sha1_git']
+ resp = self.client.get(url)
- url = reverse('browse-content',
- url_args={'query_string': non_utf8_encoded_content_data['checksums']['sha1']}) # noqa
-
- try:
- resp = self.client.get(url)
- self.assertEqual(resp.status_code, 200)
- self.assertTemplateUsed('browse/content.html')
- swh_cnt_id = get_swh_persistent_id('content', sha1_git)
- swh_cnt_id_url = reverse('browse-swh-id',
- url_args={'swh_id': swh_cnt_id})
- self.assertContains(resp, swh_cnt_id_url)
- self.assertContains(resp, escape(non_utf8_encoded_content.decode(non_utf8_encoding).encode('utf-8'))) # noqa
- except DjangoUnicodeDecodeError:
- self.fail('Textual content is not encoded in utf-8')
+ content_display = self._process_content_for_display(content)
- @patch('swh.web.browse.views.content.request_content')
- def test_content_view_image(self, mock_request_content):
+ self.assertEqual(resp.status_code, 200)
+ self.assertTemplateUsed('browse/content.html')
+ swh_cnt_id = get_swh_persistent_id('content', sha1_git)
+ swh_cnt_id_url = reverse('browse-swh-id',
+ url_args={'swh_id': swh_cnt_id})
+ self.assertContains(resp, swh_cnt_id_url)
+ self.assertContains(resp, escape(content_display['content_data']))
- mime_type = 'image/png'
- mock_request_content.return_value = stub_content_bin_data
+ @given(content_image_type())
+ def test_content_view_image(self, content):
url = reverse('browse-content',
- url_args={'query_string': stub_content_bin_data['checksums']['sha1']}) # noqa
+ url_args={'query_string': content['sha1']})
url_raw = reverse('browse-content-raw',
- url_args={'query_string': stub_content_bin_data['checksums']['sha1']}) # noqa
+ url_args={'query_string': content['sha1']})
resp = self.client.get(url)
+ content_display = self._process_content_for_display(content)
+ mimetype = content_display['mimetype']
+ content_data = content_display['content_data']
+
self.assertEqual(resp.status_code, 200)
self.assertTemplateUsed('browse/content.html')
- png_encoded = base64.b64encode(stub_content_bin_data['raw_data']) \
- .decode('utf-8')
-
self.assertContains(resp, '<img src="data:%s;base64,%s"/>'
- % (mime_type, png_encoded))
+ % (mimetype, content_data.decode('utf-8')))
self.assertContains(resp, url_raw)
- @patch('swh.web.browse.views.content.request_content')
- def test_content_view_with_path(self, mock_request_content):
- mock_request_content.return_value = stub_content_text_data
+ @given(content())
+ def test_content_view_with_path(self, content):
+
+ path = content['path']
url = reverse('browse-content',
- url_args={'query_string': stub_content_text_data['checksums']['sha1']}, # noqa
- query_params={'path': stub_content_text_path_with_root_dir}) # noqa
+ url_args={'query_string': content['sha1']},
+ query_params={'path': path})
resp = self.client.get(url)
self.assertEqual(resp.status_code, 200)
self.assertTemplateUsed('browse/content.html')
self.assertContains(resp, '<nav class="bread-crumbs')
- self.assertContains(resp, '<code class="cpp">')
- self.assertContains(resp, escape(stub_content_text_data['raw_data']))
- split_path = stub_content_text_path_with_root_dir.split('/')
+ content_display = self._process_content_for_display(content)
+ mimetype = content_display['mimetype']
+
+ if mimetype.startswith('text/'):
+ hljs_language = content['hljs-language']
+ self.assertContains(resp, '<code class="%s">' % hljs_language)
+ self.assertContains(resp, escape(content_display['content_data']))
+
+ split_path = path.split('/')
root_dir_sha1 = split_path[0]
filename = split_path[-1]
- path = stub_content_text_path_with_root_dir \
- .replace(root_dir_sha1 + '/', '') \
- .replace(filename, '')
+ path = path.replace(root_dir_sha1 + '/', '').replace(filename, '')
path_info = gen_path_info(path)
root_dir_url = reverse('browse-directory',
url_args={'sha1_git': root_dir_sha1})
self.assertContains(resp, '<li class="swh-path">',
count=len(path_info)+1)
self.assertContains(resp, '<a href="' + root_dir_url + '">' +
root_dir_sha1[:7] + '</a>')
for p in path_info:
dir_url = reverse('browse-directory',
url_args={'sha1_git': root_dir_sha1,
'path': p['path']})
self.assertContains(resp, '<a href="' + dir_url + '">' +
p['name'] + '</a>')
self.assertContains(resp, '<li>' + filename + '</li>')
url_raw = reverse('browse-content-raw',
- url_args={'query_string': stub_content_text_data['checksums']['sha1']}, # noqa
+ url_args={'query_string': content['sha1']},
query_params={'filename': filename})
self.assertContains(resp, url_raw)
url = reverse('browse-content',
- url_args={'query_string': stub_content_text_data['checksums']['sha1']}, # noqa
+ url_args={'query_string': content['sha1']},
query_params={'path': filename})
resp = self.client.get(url)
self.assertEqual(resp.status_code, 200)
self.assertTemplateUsed('browse/content.html')
self.assertNotContains(resp, '<nav class="bread-crumbs')
- @patch('swh.web.browse.views.content.request_content')
- def test_content_raw_text(self, mock_request_content):
- mock_request_content.return_value = stub_content_text_data
+ @given(content_text())
+ def test_content_raw_text(self, content):
url = reverse('browse-content-raw',
- url_args={'query_string': stub_content_text_data['checksums']['sha1']}) # noqa
+ url_args={'query_string': content['sha1']})
resp = self.client.get(url)
+ content_data = self.content_get(content['sha1'])['data']
+
self.assertEqual(resp.status_code, 200)
self.assertEqual(resp['Content-Type'], 'text/plain')
self.assertEqual(resp['Content-disposition'],
- 'filename=%s_%s' % ('sha1', stub_content_text_data['checksums']['sha1'])) # noqa
- self.assertEqual(resp.content, stub_content_text_data['raw_data'])
+ 'filename=%s_%s' % ('sha1', content['sha1']))
+ self.assertEqual(resp.content, content_data)
- filename = stub_content_text_path_with_root_dir.split('/')[-1]
+ filename = content['path'].split('/')[-1]
url = reverse('browse-content-raw',
- url_args={'query_string': stub_content_text_data['checksums']['sha1']}, # noqa
+ url_args={'query_string': content['sha1']}, # noqa
query_params={'filename': filename})
resp = self.client.get(url)
self.assertEqual(resp.status_code, 200)
self.assertEqual(resp['Content-Type'], 'text/plain')
self.assertEqual(resp['Content-disposition'],
'filename=%s' % filename)
- self.assertEqual(resp.content, stub_content_text_data['raw_data'])
-
- @patch('swh.web.browse.utils.service')
- def test_content_raw_no_utf8_text(self, mock_service):
- mock_service.lookup_content.return_value = \
- non_utf8_encoded_content_data
+ self.assertEqual(resp.content, content_data)
- mock_service.lookup_content_raw.return_value = \
- {'data': non_utf8_encoded_content}
-
- mock_service.lookup_content_filetype.return_value = None
- mock_service.lookup_content_language.return_value = None
- mock_service.lookup_content_license.return_value = None
+ @given(content_text_non_utf8())
+ def test_content_raw_no_utf8_text(self, content):
url = reverse('browse-content-raw',
- url_args={'query_string': non_utf8_encoded_content_data['checksums']['sha1']}) # noqa
+ url_args={'query_string': content['sha1']})
resp = self.client.get(url)
self.assertEqual(resp.status_code, 200)
_, encoding = get_mimetype_and_encoding_for_content(resp.content)
- self.assertEqual(encoding, non_utf8_encoding)
+ self.assertEqual(encoding, content['encoding'])
- @patch('swh.web.browse.views.content.request_content')
- def test_content_raw_bin(self, mock_request_content):
- mock_request_content.return_value = stub_content_bin_data
+ @given(content_image_type())
+ def test_content_raw_bin(self, content):
url = reverse('browse-content-raw',
- url_args={'query_string': stub_content_bin_data['checksums']['sha1']}) # noqa
+ url_args={'query_string': content['sha1']})
resp = self.client.get(url)
+ filename = content['path'].split('/')[-1]
+ content_data = self.content_get(content['sha1'])['data']
+
self.assertEqual(resp.status_code, 200)
self.assertEqual(resp['Content-Type'], 'application/octet-stream')
self.assertEqual(resp['Content-disposition'],
'attachment; filename=%s_%s' %
- ('sha1', stub_content_bin_data['checksums']['sha1']))
- self.assertEqual(resp.content, stub_content_bin_data['raw_data'])
+ ('sha1', content['sha1']))
+ self.assertEqual(resp.content, content_data)
url = reverse('browse-content-raw',
- url_args={'query_string': stub_content_bin_data['checksums']['sha1']}, # noqa
- query_params={'filename': stub_content_bin_filename})
+ url_args={'query_string': content['sha1']},
+ query_params={'filename': filename})
resp = self.client.get(url)
self.assertEqual(resp.status_code, 200)
self.assertEqual(resp['Content-Type'], 'application/octet-stream')
self.assertEqual(resp['Content-disposition'],
- 'attachment; filename=%s' % stub_content_bin_filename)
- self.assertEqual(resp.content, stub_content_bin_data['raw_data'])
+ 'attachment; filename=%s' % filename)
+ self.assertEqual(resp.content, content_data)
- @patch('swh.web.browse.views.content.request_content')
- def test_content_request_errors(self, mock_request_content):
+ @given(invalid_sha1(), unknown_content())
+ def test_content_request_errors(self, invalid_sha1, unknown_content):
- url = reverse('browse-content', url_args={'query_string': '123456'})
+ url = reverse('browse-content',
+ url_args={'query_string': invalid_sha1})
resp = self.client.get(url)
self.assertEqual(resp.status_code, 400)
self.assertTemplateUsed('error.html')
- mock_request_content.side_effect = NotFoundExc('content not found')
-
url = reverse('browse-content',
- url_args={'query_string': stub_content_text_data['checksums']['sha1']}) # noqa
+ url_args={'query_string': unknown_content['sha1']})
resp = self.client.get(url)
self.assertEqual(resp.status_code, 404)
self.assertTemplateUsed('error.html')
@patch('swh.web.browse.utils.service')
- def test_content_bytes_missing(self, mock_service):
+ @given(content())
+ def test_content_bytes_missing(self, mock_service, content):
- content_data = dict(stub_content_text_data)
- content_data['raw_data'] = None
+ content_data = self.content_get_metadata(content['sha1'])
+ content_data['data'] = None
mock_service.lookup_content.return_value = content_data
- mock_service.lookup_content_raw.side_effect = NotFoundExc('Content bytes not available!') # noqa
+ mock_service.lookup_content_raw.side_effect = NotFoundExc(
+ 'Content bytes not available!')
url = reverse('browse-content',
- url_args={'query_string': content_data['checksums']['sha1']}) # noqa
+ url_args={'query_string': content['sha1']})
resp = self.client.get(url)
+
self.assertEqual(resp.status_code, 404)
self.assertTemplateUsed('browse/content.html')
@patch('swh.web.browse.views.content.request_content')
def test_content_too_large(self, mock_request_content):
+ stub_content_too_large_data = {
+ 'checksums': {
+ 'sha1': '8624bcdae55baeef00cd11d5dfcfa60f68710a02',
+ 'sha1_git': '94a9ed024d3859793618152ea559a168bbcbb5e2',
+ 'sha256': ('8ceb4b9ee5adedde47b31e975c1d90c73ad27b6b16'
+ '5a1dcd80c7c545eb65b903'),
+ 'blake2s256': ('38702b7168c7785bfe748b51b45d9856070ba90'
+ 'f9dc6d90f2ea75d4356411ffe')
+ },
+ 'length': 3000000,
+ 'raw_data': None,
+ 'mimetype': 'text/plain',
+ 'encoding': 'us-ascii',
+ 'language': 'not detected',
+ 'licenses': 'GPL',
+ 'error_code': 200,
+ 'error_message': '',
+ 'error_description': ''
+ }
+
+ content_sha1 = stub_content_too_large_data['checksums']['sha1']
+
mock_request_content.return_value = stub_content_too_large_data
url = reverse('browse-content',
- url_args={'query_string': stub_content_too_large_data['checksums']['sha1']}) # noqa
+ url_args={'query_string': content_sha1})
url_raw = reverse('browse-content-raw',
- url_args={'query_string': stub_content_too_large_data['checksums']['sha1']}) # noqa
+ url_args={'query_string': content_sha1})
resp = self.client.get(url)
self.assertEqual(resp.status_code, 200)
self.assertTemplateUsed('browse/content.html')
self.assertContains(resp, 'Content is too large to be displayed')
self.assertContains(resp, url_raw)
+
+ def _process_content_for_display(self, content):
+ content_data = self.content_get(content['sha1'])
+
+ mime_type, encoding = get_mimetype_and_encoding_for_content(
+ content_data['data'])
+
+ mime_type, content_data = _reencode_content(mime_type, encoding,
+ content_data['data'])
+
+ return prepare_content_for_display(content_data, mime_type,
+ content['path'])
diff --git a/swh/web/tests/conftest.py b/swh/web/tests/conftest.py
index e65d042b..a70b5a42 100644
--- a/swh/web/tests/conftest.py
+++ b/swh/web/tests/conftest.py
@@ -1,53 +1,57 @@
-# Copyright (C) 2018 The Software Heritage developers
+# Copyright (C) 2018-2019 The Software Heritage developers
# See the AUTHORS file at the top-level directory of this distribution
# License: GNU Affero General Public License version 3, or any later version
# See top-level LICENSE file for more information
import json
import os
from hypothesis import settings, HealthCheck
# Register some hypothesis profiles
settings.register_profile('default', settings())
settings.register_profile(
- 'swh-web', settings(deadline=None,
- suppress_health_check=[HealthCheck.too_slow]))
+ 'swh-web',
+ settings(deadline=None,
+ suppress_health_check=[HealthCheck.too_slow,
+ HealthCheck.filter_too_much]))
settings.register_profile(
- 'swh-web-fast', settings(deadline=None, max_examples=1))
+ 'swh-web-fast',
+ settings(deadline=None, max_examples=1,
+ suppress_health_check=[HealthCheck.filter_too_much]))
def pytest_configure(config):
# Small hack in order to be able to run the unit tests
# without static assets generated by webpack.
# Those assets are not really needed for the Python tests
# but the django templates will fail to load due to missing
# generated file webpack-stats.json describing the js and css
# files to include.
# So generate a dummy webpack-stats.json file to overcome
# that issue.
test_dir = os.path.dirname(__file__)
static_dir = os.path.join(test_dir, '../static')
webpack_stats = os.path.join(static_dir, 'webpack-stats.json')
if os.path.exists(webpack_stats):
return
bundles_dir = os.path.join(test_dir, '../assets/src/bundles')
_, dirs, _ = next(os.walk(bundles_dir))
mock_webpack_stats = {
'status': 'done',
'publicPath': '/static',
'chunks': {}
}
for bundle in dirs:
asset = 'js/%s.js' % bundle
mock_webpack_stats['chunks'][bundle] = [{
'name': asset,
'publicPath': '/static/%s' % asset,
'path': os.path.join(static_dir, asset)
}]
with open(webpack_stats, 'w') as outfile:
json.dump(mock_webpack_stats, outfile)
diff --git a/swh/web/tests/data.py b/swh/web/tests/data.py
index dced201b..3204aed2 100644
--- a/swh/web/tests/data.py
+++ b/swh/web/tests/data.py
@@ -1,260 +1,276 @@
# Copyright (C) 2018-2019 The Software Heritage developers
# See the AUTHORS file at the top-level directory of this distribution
# License: GNU Affero General Public License version 3, or any later version
# See top-level LICENSE file for more information
import os
import time
from swh.indexer.language import LanguageIndexer
from swh.indexer.fossology_license import FossologyLicenseIndexer
from swh.indexer.mimetype import MimetypeIndexer
from swh.indexer.ctags import CtagsIndexer
from swh.indexer.storage import get_indexer_storage
from swh.model.hashutil import hash_to_hex, hash_to_bytes, DEFAULT_ALGORITHMS
from swh.model.identifiers import directory_identifier
from swh.loader.git.from_disk import GitLoaderFromArchive
from swh.storage.algos.dir_iterators import dir_iterator
+from swh.web.browse.utils import (
+ get_mimetype_and_encoding_for_content, prepare_content_for_display
+)
# Module used to initialize data that will be provided as tests input
# Configuration for git loader
_TEST_LOADER_CONFIG = {
'storage': {
'cls': 'memory',
'args': {}
},
'send_contents': True,
'send_directories': True,
'send_revisions': True,
'send_releases': True,
'send_snapshot': True,
'content_size_limit': 100 * 1024 * 1024,
'content_packet_size': 10,
'content_packet_size_bytes': 100 * 1024 * 1024,
'directory_packet_size': 10,
'revision_packet_size': 10,
'release_packet_size': 10,
'save_data': False,
}
# Base content indexer configuration
_TEST_INDEXER_BASE_CONFIG = {
'storage': {
'cls': 'memory',
'args': {},
},
'objstorage': {
'cls': 'memory',
'args': {},
},
'indexer_storage': {
'cls': 'memory',
'args': {},
}
}
# MimetypeIndexer with custom configuration for tests
class _MimetypeIndexer(MimetypeIndexer):
def parse_config_file(self, *args, **kwargs):
return {
**_TEST_INDEXER_BASE_CONFIG,
'tools': {
'name': 'file',
'version': '1:5.30-1+deb9u1',
'configuration': {
"type": "library",
"debian-package": "python3-magic"
}
}
}
# LanguageIndexer with custom configuration for tests
class _LanguageIndexer(LanguageIndexer):
def parse_config_file(self, *args, **kwargs):
return {
**_TEST_INDEXER_BASE_CONFIG,
'tools': {
'name': 'pygments',
'version': '2.0.1+dfsg-1.1+deb8u1',
'configuration': {
'type': 'library',
'debian-package': 'python3-pygments',
'max_content_size': 10240,
}
}
}
# FossologyLicenseIndexer with custom configuration for tests
class _FossologyLicenseIndexer(FossologyLicenseIndexer):
def parse_config_file(self, *args, **kwargs):
return {
**_TEST_INDEXER_BASE_CONFIG,
'workdir': '/tmp/swh/indexer.fossology.license',
'tools': {
'name': 'nomos',
'version': '3.1.0rc2-31-ga2cbb8c',
'configuration': {
'command_line': 'nomossa <filepath>',
},
}
}
# CtagsIndexer with custom configuration for tests
class _CtagsIndexer(CtagsIndexer):
def parse_config_file(self, *args, **kwargs):
return {
**_TEST_INDEXER_BASE_CONFIG,
'workdir': '/tmp/swh/indexer.ctags',
'languages': {'c': 'c'},
'tools': {
'name': 'universal-ctags',
'version': '~git7859817b',
'configuration': {
'command_line': '''ctags --fields=+lnz --sort=no --links=no ''' # noqa
'''--output-format=json <filepath>'''
},
}
}
# Lightweight git repositories that will be loaded to generate
# input data for tests
_TEST_ORIGINS = [
{
'id': 1,
'type': 'git',
'url': 'https://github.com/wcoder/highlightjs-line-numbers.js',
'archives': ['highlightjs-line-numbers.js.zip',
'highlightjs-line-numbers.js_visit2.zip']
},
{
'id': 2,
'type': 'git',
'url': 'https://github.com/memononen/libtess2',
'archives': ['libtess2.zip']
},
{
'id': 3,
'type': 'git',
'url': 'repo_with_submodules',
'archives': ['repo_with_submodules.tgz']
}
]
# Tests data initialization
def _init_tests_data():
# Load git repositories from archives
loader = GitLoaderFromArchive(config=_TEST_LOADER_CONFIG)
for origin in _TEST_ORIGINS:
nb_visits = len(origin['archives'])
for i, archive in enumerate(origin['archives']):
origin_repo_archive = \
os.path.join(os.path.dirname(__file__),
'resources/repos/%s' % archive)
loader.load(origin['url'], origin_repo_archive, None)
if nb_visits > 1 and i != nb_visits - 1:
time.sleep(1)
# Get reference to the memory storage
storage = loader.storage
contents = set()
directories = set()
revisions = set()
releases = set()
snapshots = set()
persons = set()
+ content_path = {}
+
# Get all objects loaded into the test archive
for origin in _TEST_ORIGINS:
snp = storage.snapshot_get_latest(origin['id'])
snapshots.add(hash_to_hex(snp['id']))
for branch_name, branch_data in snp['branches'].items():
if branch_data['target_type'] == 'revision':
revisions.add(branch_data['target'])
elif branch_data['target_type'] == 'release':
release = next(storage.release_get([branch_data['target']]))
revisions.add(release['target'])
releases.add(hash_to_hex(branch_data['target']))
persons.add(release['author']['id'])
for rev_log in storage.revision_shortlog(set(revisions)):
rev_id = rev_log[0]
revisions.add(rev_id)
for rev in storage.revision_get(revisions):
dir_id = rev['directory']
persons.add(rev['author']['id'])
persons.add(rev['committer']['id'])
directories.add(hash_to_hex(dir_id))
for entry in dir_iterator(storage, dir_id):
+ content_path[entry['sha1']] = '/'.join(
+ [hash_to_hex(dir_id), entry['path'].decode('utf-8')])
if entry['type'] == 'file':
contents.add(entry['sha1'])
elif entry['type'] == 'dir':
directories.add(hash_to_hex(entry['target']))
# Get all checksums for each content
contents_metadata = storage.content_get_metadata(contents)
contents = []
for content_metadata in contents_metadata:
contents.append({
algo: hash_to_hex(content_metadata[algo])
for algo in DEFAULT_ALGORITHMS
})
+ path = content_path[content_metadata['sha1']]
+ cnt = next(storage.content_get([content_metadata['sha1']]))
+ mimetype, encoding = get_mimetype_and_encoding_for_content(cnt['data'])
+ content_display_data = prepare_content_for_display(
+ cnt['data'], mimetype, path)
+ contents[-1]['path'] = path
+ contents[-1]['mimetype'] = mimetype
+ contents[-1]['encoding'] = encoding
+ contents[-1]['hljs-language'] = content_display_data['language']
# Create indexer storage instance that will be shared by indexers
idx_storage = get_indexer_storage('memory', {})
# Instantiate content indexers that will be used in tests
# and force them to use the memory storages
indexers = {}
for idx_name, idx_class in (('mimetype_indexer', _MimetypeIndexer),
('language_indexer', _LanguageIndexer),
('license_indexer', _FossologyLicenseIndexer),
('ctags_indexer', _CtagsIndexer)):
idx = idx_class()
idx.storage = storage
idx.objstorage = storage.objstorage
idx.idx_storage = idx_storage
idx.register_tools(idx.config['tools'])
indexers[idx_name] = idx
# Add the empty directory to the test archive
empty_dir_id = directory_identifier({'entries': []})
empty_dir_id_bin = hash_to_bytes(empty_dir_id)
storage.directory_add([{'id': empty_dir_id_bin, 'entries': []}])
# Return tests data
return {
'storage': storage,
'idx_storage': idx_storage,
**indexers,
'origins': _TEST_ORIGINS,
'contents': contents,
'directories': list(directories),
'persons': list(persons),
'releases': list(releases),
'revisions': list(map(hash_to_hex, revisions)),
'snapshots': list(snapshots)
}
_tests_data = None
def get_tests_data():
"""
Initialize tests data and return them in a dict.
"""
global _tests_data
if _tests_data is None:
_tests_data = _init_tests_data()
return _tests_data
diff --git a/swh/web/tests/strategies.py b/swh/web/tests/strategies.py
index 8b79dfcd..e8087a6d 100644
--- a/swh/web/tests/strategies.py
+++ b/swh/web/tests/strategies.py
@@ -1,467 +1,501 @@
# Copyright (C) 2018-2019 The Software Heritage developers
# See the AUTHORS file at the top-level directory of this distribution
# License: GNU Affero General Public License version 3, or any later version
# See top-level LICENSE file for more information
import random
from collections import defaultdict
from datetime import datetime
from hypothesis import settings, assume
from hypothesis.strategies import (
just, sampled_from, lists, composite, datetimes,
integers, binary, text, characters
)
from swh.model.hashutil import hash_to_hex, hash_to_bytes
from swh.model.identifiers import directory_identifier
from swh.storage.algos.revisions_walker import get_revisions_walker
from swh.storage.tests.algos.test_snapshot import ( # noqa
origins as new_origin_strategy, snapshots as new_snapshot
)
from swh.web.tests.data import get_tests_data
# Module dedicated to the generation of input data for tests through
# the use of hypothesis.
# Some of these data are sampled from a test archive created and populated
# in the swh.web.tests.data module.
# Set the swh-web hypothesis profile if none has been explicitly set
hypothesis_default_settings = settings.get_profile('default')
if repr(settings()) == repr(hypothesis_default_settings):
settings.load_profile('swh-web')
# Import tests data
tests_data = get_tests_data()
storage = tests_data['storage']
# The following strategies exploit the hypothesis capabilities
def _known_swh_object(object_type):
return sampled_from(tests_data[object_type])
def sha1():
"""
Hypothesis strategy returning a valid hexadecimal sha1 value.
"""
return binary(
min_size=20, max_size=20).filter(
lambda s: int.from_bytes(s, byteorder='little')).map(hash_to_hex)
def invalid_sha1():
"""
Hypothesis strategy returning an invalid sha1 representation.
"""
return binary(
min_size=50, max_size=50).filter(
lambda s: int.from_bytes(s, byteorder='little')).map(hash_to_hex)
def sha256():
"""
Hypothesis strategy returning a valid hexadecimal sha256 value.
"""
return binary(
min_size=32, max_size=32).filter(
lambda s: int.from_bytes(s, byteorder='little')).map(hash_to_hex)
def content():
"""
Hypothesis strategy returning a random content ingested
into the test archive.
"""
return _known_swh_object('contents')
def contents():
"""
Hypothesis strategy returning random contents ingested
into the test archive.
"""
return lists(content(), min_size=2, max_size=8)
+def content_text():
+ """
+ Hypothesis strategy returning random textual contents ingested
+ into the test archive.
+ """
+ return content().filter(lambda c: c['mimetype'].startswith('text/'))
+
+
+def content_text_non_utf8():
+ """
+ Hypothesis strategy returning random textual contents not encoded
+ to UTF-8 ingested into the test archive.
+ """
+ return content().filter(lambda c: c['mimetype'].startswith('text/') and
+ c['encoding'] not in ('utf-8', 'us-ascii'))
+
+
+def content_text_no_highlight():
+ """
+ Hypothesis strategy returning random textual contents with no detected
+ programming language to highlight ingested into the test archive.
+ """
+ return content().filter(lambda c: c['mimetype'].startswith('text/') and
+ c['hljs-language'] == 'nohighlight')
+
+
+def content_image_type():
+ """
+ Hypothesis strategy returning random image contents ingested
+ into the test archive.
+ """
+ return content().filter(lambda c: c['mimetype'].startswith('image/'))
+
+
@composite
def new_content(draw):
blake2s256_hex = draw(sha256())
sha1_hex = draw(sha1())
sha1_git_hex = draw(sha1())
sha256_hex = draw(sha256())
assume(sha1_hex != sha1_git_hex)
assume(blake2s256_hex != sha256_hex)
return {
'blake2S256': blake2s256_hex,
'sha1': sha1_hex,
'sha1_git': sha1_git_hex,
'sha256': sha256_hex
}
def unknown_content():
"""
Hypothesis strategy returning a random content not ingested
into the test archive.
"""
return new_content().filter(
lambda c: next(storage.content_get(
[hash_to_bytes(c['sha1'])])) is None)
def unknown_contents():
"""
Hypothesis strategy returning random contents not ingested
into the test archive.
"""
return lists(unknown_content(), min_size=2, max_size=8)
def directory():
"""
Hypothesis strategy returning a random directory ingested
into the test archive.
"""
return _known_swh_object('directories')
def empty_directory():
"""
Hypothesis strategy returning the empty directory ingested
into the test archive.
"""
return just(directory_identifier({'entries': []}))
def unknown_directory():
"""
Hypothesis strategy returning a random directory not ingested
into the test archive.
"""
return sha1().filter(
lambda s: len(list(storage.directory_missing([hash_to_bytes(s)]))) > 0)
def origin():
"""
Hypothesis strategy returning a random origin ingested
into the test archive.
"""
return _known_swh_object('origins')
def origin_with_multiple_visits():
"""
Hypothesis strategy returning a random origin ingested
into the test archive.
"""
ret = []
for origin in tests_data['origins']:
visits = list(storage.origin_visit_get(origin['id']))
if len(visits) > 1:
ret.append(origin)
return sampled_from(ret)
def unknown_origin_id():
"""
Hypothesis strategy returning a random origin id not ingested
into the test archive.
"""
return integers(min_value=1000000)
def new_origin():
"""
Hypothesis strategy returning a random origin not ingested
into the test archive.
"""
return new_origin_strategy().filter(
lambda origin: storage.origin_get(origin) is None)
def new_origins(nb_origins=None):
"""
Hypothesis strategy returning random origins not ingested
into the test archive.
"""
min_size = nb_origins if nb_origins is not None else 2
max_size = nb_origins if nb_origins is not None else 8
size = random.randint(min_size, max_size)
return lists(new_origin(), min_size=size, max_size=size,
unique_by=lambda o: tuple(sorted(o.items())))
def visit_dates(nb_dates=None):
"""
Hypothesis strategy returning a list of visit dates.
"""
min_size = nb_dates if nb_dates else 2
max_size = nb_dates if nb_dates else 8
return lists(datetimes(min_value=datetime(2015, 1, 1, 0, 0),
max_value=datetime(2018, 12, 31, 0, 0)),
min_size=min_size, max_size=max_size, unique=True).map(sorted)
def release():
"""
Hypothesis strategy returning a random release ingested
into the test archive.
"""
return _known_swh_object('releases')
def unknown_release():
"""
Hypothesis strategy returning a random revision not ingested
into the test archive.
"""
return sha1().filter(
lambda s: next(storage.release_get([s])) is None)
def revision():
"""
Hypothesis strategy returning a random revision ingested
into the test archive.
"""
return _known_swh_object('revisions')
def unknown_revision():
"""
Hypothesis strategy returning a random revision not ingested
into the test archive.
"""
return sha1().filter(
lambda s: next(storage.revision_get([hash_to_bytes(s)])) is None)
@composite
def new_person(draw):
"""
Hypothesis strategy returning random raw swh person data.
"""
name = draw(text(min_size=5, max_size=30,
alphabet=characters(min_codepoint=0, max_codepoint=255)))
email = '%s@company.org' % name
return {
'name': name.encode(),
'email': email.encode(),
'fullname': ('%s <%s>' % (name, email)).encode()
}
@composite
def new_swh_date(draw):
"""
Hypothesis strategy returning random raw swh date data.
"""
timestamp = draw(
datetimes(min_value=datetime(2015, 1, 1, 0, 0),
max_value=datetime(2018, 12, 31, 0, 0)).map(
lambda d: int(d.timestamp())))
return {
'timestamp': timestamp,
'offset': 0,
'negative_utc': False,
}
@composite
def new_revision(draw):
"""
Hypothesis strategy returning random raw swh revision data
not ingested into the test archive.
"""
return {
'id': draw(unknown_revision().map(hash_to_bytes)),
'directory': draw(sha1().map(hash_to_bytes)),
'author': draw(new_person()),
'committer': draw(new_person()),
'message': draw(
text(min_size=20, max_size=100).map(lambda t: t.encode())),
'date': draw(new_swh_date()),
'committer_date': draw(new_swh_date()),
'synthetic': False,
'type': 'git',
'parents': [],
'metadata': [],
}
def revisions():
"""
Hypothesis strategy returning random revisions ingested
into the test archive.
"""
return lists(revision(), min_size=2, max_size=8)
def unknown_revisions():
"""
Hypothesis strategy returning random revisions not ingested
into the test archive.
"""
return lists(unknown_revision(), min_size=2, max_size=8)
def snapshot():
"""
Hypothesis strategy returning a random snapshot ingested
into the test archive.
"""
return _known_swh_object('snapshots')
def new_snapshots(nb_snapshots=None):
min_size = nb_snapshots if nb_snapshots else 2
max_size = nb_snapshots if nb_snapshots else 8
return lists(new_snapshot(min_size=2, max_size=10, only_objects=True),
min_size=min_size, max_size=max_size)
def unknown_snapshot():
"""
Hypothesis strategy returning a random revision not ingested
into the test archive.
"""
return sha1().filter(
lambda s: storage.snapshot_get(hash_to_bytes(s)) is None)
def person():
"""
Hypothesis strategy returning a random person ingested
into the test archive.
"""
return _known_swh_object('persons')
def unknown_person():
"""
Hypothesis strategy returning a random person not ingested
into the test archive.
"""
persons = tests_data['persons']
return integers(min_value=max(persons)+1)
def _get_origin_dfs_revisions_walker():
origin = random.choice(tests_data['origins'][:-1])
snapshot = storage.snapshot_get_latest(origin['id'])
head = snapshot['branches'][b'HEAD']['target']
return get_revisions_walker('dfs', storage, head)
def ancestor_revisions():
"""
Hypothesis strategy returning a pair of revisions ingested into the
test archive with an ancestor relation.
"""
# get a dfs revisions walker for one of the origins
# loaded into the test archive
revisions_walker = _get_origin_dfs_revisions_walker()
master_revisions = []
children = defaultdict(list)
init_rev_found = False
# get revisions only authored in the master branch
for rev in revisions_walker:
for rev_p in rev['parents']:
children[rev_p].append(rev['id'])
if not init_rev_found:
master_revisions.append(rev)
if not rev['parents']:
init_rev_found = True
# head revision
root_rev = master_revisions[0]
# pick a random revision, different from head, only authored
# in the master branch
ancestor_rev_idx = random.choice(list(range(1, len(master_revisions)-1)))
ancestor_rev = master_revisions[ancestor_rev_idx]
ancestor_child_revs = children[ancestor_rev['id']]
return just({
'sha1_git_root': hash_to_hex(root_rev['id']),
'sha1_git': hash_to_hex(ancestor_rev['id']),
'children': [hash_to_hex(r) for r in ancestor_child_revs]
})
def non_ancestor_revisions():
"""
Hypothesis strategy returning a pair of revisions ingested into the
test archive with no ancestor relation.
"""
# get a dfs revisions walker for one of the origins
# loaded into the test archive
revisions_walker = _get_origin_dfs_revisions_walker()
merge_revs = []
children = defaultdict(list)
# get all merge revisions
for rev in revisions_walker:
if len(rev['parents']) > 1:
merge_revs.append(rev)
for rev_p in rev['parents']:
children[rev_p].append(rev['id'])
# find a merge revisions whose parents have a unique child revision
random.shuffle(merge_revs)
selected_revs = None
for merge_rev in merge_revs:
if all(len(children[rev_p]) == 1
for rev_p in merge_rev['parents']):
selected_revs = merge_rev['parents']
return just({
'sha1_git_root': hash_to_hex(selected_revs[0]),
'sha1_git': hash_to_hex(selected_revs[1])
})
# The following strategies returns data specific to some tests
# that can not be generated and thus are hardcoded.
def contents_with_ctags():
"""
Hypothesis strategy returning contents ingested into the test
archive. Those contents are ctags compatible, that is running
ctags on those lay results.
"""
return just({
'sha1s': ['0ab37c02043ebff946c1937523f60aadd0844351',
'15554cf7608dde6bfefac7e3d525596343a85b6f',
'2ce837f1489bdfb8faf3ebcc7e72421b5bea83bd',
'30acd0b47fc25e159e27a980102ddb1c4bea0b95',
'4f81f05aaea3efb981f9d90144f746d6b682285b',
'5153aa4b6e4455a62525bc4de38ed0ff6e7dd682',
'59d08bafa6a749110dfb65ba43a61963d5a5bf9f',
'7568285b2d7f31ae483ae71617bd3db873deaa2c',
'7ed3ee8e94ac52ba983dd7690bdc9ab7618247b4',
'8ed7ef2e7ff9ed845e10259d08e4145f1b3b5b03',
'9b3557f1ab4111c8607a4f2ea3c1e53c6992916c',
'9c20da07ed14dc4fcd3ca2b055af99b2598d8bdd',
'c20ceebd6ec6f7a19b5c3aebc512a12fbdc9234b',
'e89e55a12def4cd54d5bff58378a3b5119878eb7',
'e8c0654fe2d75ecd7e0b01bee8a8fc60a130097e',
'eb6595e559a1d34a2b41e8d4835e0e4f98a5d2b5'],
'symbol_name': 'ABS'
})
def revision_with_submodules():
"""
Hypothesis strategy returning a revision that is known to
point to a directory with revision entries (aka git submodule)
"""
return just({
'rev_sha1_git': 'ffcb69001f3f6745dfd5b48f72ab6addb560e234',
'rev_dir_sha1_git': 'd92a21446387fa28410e5a74379c934298f39ae2',
'rev_dir_rev_path': 'libtess2'
})